package masterserver;


import java.awt.List;
import java.io.EOFException;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.lang.management.ManagementFactory;
import java.lang.management.ThreadMXBean;
import java.net.ServerSocket;
import java.net.Socket;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.HashMap;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.Observable;
import java.util.Observer;
import java.util.Timer;
import java.util.TimerTask;
import java.util.TreeSet;
import java.util.logging.Logger;

import org.w3c.dom.ls.LSInput;


import result.Result;

import client.ClientDescription;
import comm.ClientPriority;
import comm.ConnectionType;
import comm.InfrastructureInformation;
import comm.SystemInformation;
import stats.Stat;
import tasks.Task;
import tasks.TaskDescription;
import tasks.TaskId;
import worker.ClassLoaderObjectInputStream;
import worker.WorkerData;

public class MasterServer {

	//private TreeSet<TaskDescription> pendingTasks = new TreeSet<TaskDescription>();
	LinkedList<LinkedList<TaskDescription>> pendingTasks = new LinkedList<LinkedList<TaskDescription>>();
	private LinkedList<ClientDescription> canceledTaskList = new LinkedList<ClientDescription>();	
	private Hashtable<Integer, WorkerData> workerDataVector = new Hashtable<Integer, WorkerData>();

	private int workerID = 0;
	private int taskCounter = 0;

	private TalkToCodeServer talkToCodeServer;

	private final static String className=MasterServer.class.getName();
	private static Logger log=Logger.getLogger(className);

	private Stat statistics = new Stat();
	private long startTime;
	private int clientID=0;
	private static ThreadMXBean tmxb = ManagementFactory.getThreadMXBean();
	private InfrastructureInformation infrastructureInformation;

	private long numberOfCores = 0;
	private long numberOfWorkers = 0;
	private canceledTaskObserver taskObserver = new canceledTaskObserver();
	public MasterServer(InfrastructureInformation infrastructureInformation) {
		super();
		this.infrastructureInformation = infrastructureInformation;

		startTime = System.currentTimeMillis();

		System.out.println(tmxb.getCurrentThreadCpuTime());
		System.out.println(tmxb.getCurrentThreadUserTime());
		System.out.println(tmxb.getDaemonThreadCount());
	}

	public void execute() {
		connectCodeServer();
		new ReceiveConnectionsThread().start();
	}

	private void connectCodeServer() {

		talkToCodeServer = new TalkToCodeServer(infrastructureInformation.getCodeServerIPaddress(), infrastructureInformation.getCodeServerEncryptedPort());
		talkToCodeServer.connectCodeServer();

	}
	private synchronized int addWorkerData(WorkerData data){

		workerDataVector.put(this.workerID, data);
		
		return workerID++;
	}

	private synchronized void removeWorker(int id){

		workerDataVector.remove(id);

	}

	private synchronized void addTaskList(LinkedList<TaskDescription> taskList) {

		pendingTasks.add(taskList);
		System.out.println("Number of lists:"+pendingTasks.size());
		notifyAll();
	}
	private synchronized void addSingleTask(TaskDescription task) {
		boolean flag = false;

			
		
		Iterator<LinkedList<TaskDescription>> taskIterator = pendingTasks.iterator();
		while(taskIterator.hasNext()){
			LinkedList<TaskDescription> list = (LinkedList<TaskDescription>)taskIterator.next();
			Iterator<TaskDescription> tl = list.iterator();
			while(tl.hasNext()){
				TaskDescription td =(TaskDescription) tl.next();
				if(td.getClientID()==task.getClientID())
				{
					list.add(task);
					flag = true;
					break;
				}
		}
		}
		if(!flag){
			 LinkedList<TaskDescription> localTaskList = new LinkedList<TaskDescription>();
			 localTaskList.add(task);
			 addTaskList(localTaskList);
			 notifyAll();		}

		System.out.println("Number of tasks:"+pendingTasks.size());

	}

	private synchronized void masterStat(int numberOfCores){
		this.numberOfCores += numberOfCores;
		this.numberOfWorkers++;
		System.out.println("### Number of cores:"+this.numberOfCores+" Number of workers:"+this.numberOfWorkers);
	}

	private synchronized TaskDescription getTask(WorkerData workerData) {
		TaskDescription task = null;
		System.out.println("enter");
		
		try {
//Bug here ... when workers are waiting for tasks and no tasks are at the queue only one task is dispatched when available
			while (pendingTasks.size() == 0) {
				System.out.println("Waiting");
				wait();
			}

		} catch (InterruptedException e) {
			e.printStackTrace();
			System.out.println(getIdAndTime(workerData)
					+ " ->Was interrupted while waiting!");
			return null;
		}

		int maxPriority=-1;
		int id;

		LinkedList<TaskDescription> taskListToReturn = null;
		TaskDescription taskToReturn = null;
		
				Iterator<LinkedList<TaskDescription>> clientIterator = pendingTasks.iterator();
			while(clientIterator.hasNext()){
				LinkedList<TaskDescription> list = (LinkedList<TaskDescription>)clientIterator.next();
				Iterator<TaskDescription> tl = list.iterator();
				while(tl.hasNext()){
					TaskDescription td =(TaskDescription) tl.next();
					if(td.getPriority().ordinal()>maxPriority)
					{
						taskListToReturn =list;

					}
				}
			}
			taskToReturn = taskListToReturn.pollFirst();
			Iterator<LinkedList<TaskDescription>> clientIterator2 = pendingTasks.iterator();
			while(clientIterator2.hasNext()){
				LinkedList<TaskDescription> list = (LinkedList<TaskDescription>)clientIterator2.next();
				if(list.size()==0)
					clientIterator2.remove();
			}

		
		if (taskToReturn == null) {
			System.out.println("XXX: got null task from pending list!");
		}

	System.out.println("New task sent...");
		return taskToReturn;
	}

	class canceledTaskObserver extends Observable{
		private ClientDescription clientDescription;
		public void canceledTaskWarning(ClientDescription clientDescription){
			this.clientDescription = clientDescription;
			
			synchronized (pendingTasks) {

				Iterator<LinkedList<TaskDescription>> clientIterator2 = pendingTasks.iterator();
				while(clientIterator2.hasNext()){
					LinkedList<TaskDescription> list = (LinkedList<TaskDescription>)clientIterator2.next();
					
					Iterator<TaskDescription> tl = list.iterator();
				
					while(tl.hasNext()){
						TaskDescription td =(TaskDescription) tl.next();
						if(td.getClientID() == clientDescription.getID())
						{
							list.remove();
							clientIterator2.remove();
							break;
							
						}
					
				}
					
				
				
				}
			}
		
			setChanged();
			notifyObservers(clientDescription);
		
			
		
	}
	}

	
	private String getIdAndTime(WorkerData workerData) {
		return "(" + workerData.getId() + ") ["
		+ (System.currentTimeMillis() - startTime) + "]";
	}


	class ReceiveConnectionsThread extends Thread {
		public ReceiveConnectionsThread() {

		}

		public void run() {
			ServerSocket serverSocket = null;
			try {
				serverSocket = new ServerSocket(infrastructureInformation.getMasterServerEncryptedPort());
				System.out.println("Waiting for connections on socket: "
						+ serverSocket);
				while (true) {
					try {
						Socket socket = serverSocket.accept();
						socket.setSoTimeout(0);
						socket.setKeepAlive(true);

						ObjectOutputStream out = new ObjectOutputStream(
								socket.getOutputStream());

						System.out.println("new connection");
						ClassLoaderObjectInputStream in = new ClassLoaderObjectInputStream(
								socket.getInputStream(),talkToCodeServer);


						ConnectionType type = (ConnectionType) in.readObject();
						System.out.println(type);
						switch (type) {
						case CLIENT:
							new ClientThread(socket, in, out).start();
							break;
						case WORKER:
							WorkerThread wt = new  WorkerThread(socket, in, out);
							taskObserver.addObserver(wt);
							wt.start();
							break;
						default:
							System.out.println("ERROR!");
						}

					} catch (IOException e) {

					} catch (ClassNotFoundException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
				}
			} catch (IOException e) {
				e.printStackTrace();
			} finally {
				try {
					if (serverSocket != null)
						serverSocket.close();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}
	}

	class ClientThread extends Thread {
		private Socket socket;
		private ObjectOutputStream out;
		private ClassLoaderObjectInputStream in;
		private int problemNumber;
		private int version;
		private ClientPriority priority;
		private LinkedList<TaskDescription> localTaskList = new LinkedList<TaskDescription>();

		private ClientDescription description;

		
		public ClientThread(Socket socket, ClassLoaderObjectInputStream in,
				ObjectOutputStream out) {
			this.socket = socket;
			this.in = in;
			this.out = out;

			System.out.println("New Client " + socket);
		}

		public void run() {
			try {
				while (true) {
					ConnectionType type = (ConnectionType) in.readObject();


					switch (type) {
					case CLIENT_NEW_PROBLEM:
						try {
							problemNumber = (Integer) in.readObject();
							version = (Integer) in.readObject();
							in.setProblemAndVersion(problemNumber, version);
							this.priority =  (ClientPriority) in.readObject();
							description = new ClientDescription(clientID++,priority,out,problemNumber,version);

						} catch (IOException e) {
							// TODO Auto-generated catch block
							e.printStackTrace();
						} catch (ClassNotFoundException e) {
							// TODO Auto-generated catch block
							e.printStackTrace();
						}

						break;
					case CLIENT_NEW_TASK:
						System.out.println("Waiting for new task!");
						Task task = (Task) in.readObject();
						System.out.println("got new Task:" + task);
						statistics.setInputedTasks(); //1 more task inputed
						localTaskList.add(new TaskDescription(task, description,taskCounter++));
						//addTask(new TaskDescription(task, description,taskCounter++));
						//System.out.println(task);
						break;
					case CLIENT_CANCEL_ALL_TASKS:
						System.out.println(description.toString());
						taskObserver.canceledTaskWarning(description);
						break;
					case START_WORK:
						addTaskList(localTaskList);
						break;
					}
				}
			} catch (IOException e) {
				e.printStackTrace();
				taskObserver.canceledTaskWarning(description);
				
					try {
						in.close();
						out.close();
						socket.close();
					} catch (IOException e1) {
						// TODO Auto-generated catch block
						e1.printStackTrace();
					}
					
				return;
			} catch (ClassNotFoundException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}

		}

	}

	private class WorkerThread extends Thread implements Observer {
		private Socket socket;
		private ObjectOutputStream out;
		private ClassLoaderObjectInputStream in;
		private WorkerData workerData;
		private SystemInformation si;
		private int workerID;
		private LinkedList<TaskDescription> taskList = new LinkedList<TaskDescription>();

		private PingPong pingpong = new PingPong();
		private boolean socketInFlag = false;
		
		public WorkerThread(Socket socket, ClassLoaderObjectInputStream in,
				ObjectOutputStream out) {
			this.socket = socket;
			this.in = in;
			this.out = out;
			System.out.println("New slave " + socket);
			workerData = new WorkerData();
		}

		public void run() {

			try {
				
				workerData = (WorkerData)in.readObject();
				workerData.setWorkerAddress(socket.getInetAddress().toString());
				workerData.setWorkerPort(socket.getPort());
				workerData.setId(workerID);

				this.workerID = addWorkerData(workerData);

				out.writeObject(workerData);
				masterStat(workerData.getNumberOfProcessors());
				System.out.println(workerData.toString());

				pingpong.start();

				ConnectionType type;

				while (true) {
					type = (ConnectionType)in.readObject();
					//System.out.println(type);
					switch(type){
					case WORKER_FEED:

						new feedWorker().start();
						break;
					case WORKER_RESULTS:
						socketInFlag = true;
						TaskId tid = (TaskId) in.readObject();
						in.setProblemAndVersion(tid.getProblemNumber(), tid.getVersion());
						Result result = (Result) in.readObject();
						workerData = (WorkerData) in.readObject();
						
						System.out.println("recebi resultados"+result.toString());
						Iterator<TaskDescription> taskIterator = taskList.iterator();
						while(taskIterator.hasNext()){
							TaskDescription task = (TaskDescription)taskIterator.next();
							//System.out.println(task.getTaskId()+"=="+tid.getTaskId());
							if(task.getTaskId() == tid.getTaskId()){
								task.getClientOutStream().writeObject(result);
								synchronized (taskList) {
									taskList.remove(task);	
								}
		
								break;
							}
	
						}
						socketInFlag = false;
						break;
					case WORKER_CANCELED_TASKS:
						
						System.out.println("O worker avisou que matou tasks");
						break;

					case WORKER_CANCELED_TASK_OK:
						int taskID = (Integer) in.readObject();
						System.out.println(" KILL OK "+taskID);
						synchronized (taskList) {
							Iterator<TaskDescription> taskIteratorDel = taskList.iterator();
							
							while(taskIteratorDel.hasNext()){
								TaskDescription task = (TaskDescription)taskIteratorDel.next();
								if(task.getTaskId() == taskID){
									System.out.println("Trying to remove: "+task.getTaskId());
							     	taskIteratorDel.remove();
									taskList.remove(task);
								}else
									System.out.println("not found...");
							}
						}
						
						break;

					case PONG:
						pingpong.acknolegde();
						break;
					}
				}

			} 
			catch (EOFException e) {
				System.out.println("EX:"+e);
				destroyWorkerSaveTask();
				return;
			}
			catch (IOException e) {
				// TODO Auto-generated catch block
				System.out.println("EX1");
				e.printStackTrace();
				destroyWorkerSaveTask();
				return;
			} catch (ClassNotFoundException e) {
				System.out.println(workerID+" EX2");
				e.printStackTrace();

			}


		}
		private class PingPong extends Thread{
			private int flag = 0;
			private final static int TIME_OUT=5000;
			private static final int TIME_TO_SEND_NEXT_PING = 7000;
			public PingPong(){

			}

			public synchronized void waiter(){

				try {
					while(flag==0 && !socketInFlag){
						wait(TIME_OUT);
						if(flag == 0 && !socketInFlag)
							flag = 2;
						return;
					}
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}

			public synchronized void acknolegde(){
				flag = 1;
				notify();

			}

			public void run(){

				while(true){
					try {
						Thread.sleep(TIME_TO_SEND_NEXT_PING);
					} catch (InterruptedException e1) {
						// TODO Auto-generated catch block
						e1.printStackTrace();
					}
					synchronized (out) {
						try {
							out.writeObject(ConnectionType.PING);
						} catch (IOException e) {
							// TODO Auto-generated catch block
							e.printStackTrace();
							destroyWorkerSaveTask();
							return;	
						}
					}
					waiter();
					if(flag==1 || socketInFlag){
						flag = 0;
					}
					else{
						System.out.println("Timeout");
						destroyWorkerSaveTask();
						return;
					}

				}
			}
		}

		public void destroyWorkerSaveTask(){
			try {
				in.close();
				out.close();
				socket.close();

				Iterator<TaskDescription> taskIterator = taskList.iterator();
				while(taskIterator.hasNext()){
					TaskDescription task = (TaskDescription)taskIterator.next();
					addSingleTask(task);
				}
				removeWorker(this.workerID);
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}

		private class feedWorker extends Thread{
			public feedWorker(){

			}
			public void run(){
				
				System.out.println("MasterServer: SLAVE PROC INFO:"+workerData.getNumberOfProcessors()); //we have number of available cores
				TaskDescription taskDescription = getTask(workerData);
				
				synchronized (taskList) {

					taskList.add(taskDescription);
				}
				System.out.println("Ready to send new task to slave!!! " + taskDescription.getTask());
				try {
				
					synchronized (out) {
						out.writeObject(ConnectionType.FEED_WORKER);
						out.writeObject(new TaskId(taskDescription.getClientID(),taskDescription.getVersion(),taskDescription.getProblemNumber(),taskDescription.getTaskId()));
						System.out.println("Sending task ... ");					
						out.writeObject(taskDescription.getTask());
					}
					
					System.out.println("Task Sent...");
					statistics.setTasksServed();

				} catch (IOException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
					destroyWorkerSaveTask();
					return;
				}
			}
		}

@Override
public void update(Observable o, Object arg) {
	System.out.println("Cancel worker"+arg.toString());
	ClientDescription clientDescription = (ClientDescription) arg;

		
				synchronized (taskList) {
			Iterator<TaskDescription> taskIterator = taskList.iterator();
			while(taskIterator.hasNext()){
				TaskDescription task = (TaskDescription)taskIterator.next();
				if(task.getClientID() == clientDescription.getID()){
					taskIterator.remove();
					synchronized (out) {
						try {
						
							out.writeObject(ConnectionType.CLIENT_CANCEL_ALL_TASKS);
							System.out.println("kill"+task.getTaskId());
							out.writeObject(task.getTaskId());
						} catch (IOException e) {
							// TODO Auto-generated catch block
							e.printStackTrace();
							destroyWorkerSaveTask();
							return;
						}

					}
								
				}
				
			}

}
	}

	class StatisticsThread extends Thread {


		public StatisticsThread() {

		}

		public void run() {
			int count = 0;


			while (true) {
				System.out.println("****************** STATS *******************");
				System.out.println("STATS n: +"+ count++ +" : "+statistics.toString());
				System.out.println("****************** END STATS *******************");
				try {
					Thread.sleep(10000);
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}


		}
	}


	}
}

